package belf.migrate.engine.job.execute.helper;

import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.table.schema.TablePath;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.job.Job;
import belf.migrate.engine.job.model.execute.ExecuteResult;
import belf.migrate.engine.job.model.execute.ExecuteSqlResult;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

@Slf4j
public class ExecuteSqlTool {

    private Job job;

    private JobContext jobContext;

    private AbstractSinkCatalog catalog;

    public ExecuteSqlTool(JobContext jobContext, Job job) {
        this.jobContext = jobContext;
        this.catalog = jobContext.getSinkCatalog();
        this.job = job;
    }

    public ExecuteResult run() {

        String sqls = job.getMissions().get(0);
        String[] sqlList = sqls.split(";");
        ExecuteResult executeResult = new ExecuteResult();
        String databaseName = job.getTask().getTaskConf().getSinkConf().getDatabaseName();
        String schemaName = job.getTask().getTaskConf().getSinkConf().getSchemaName();
        QualifiersModel qualifiersModel = jobContext.getSourceQualifiersModel();

        TablePath tablePath = TablePath.of(databaseName, schemaName, null, qualifiersModel);

        try {
            List<ExecuteSqlResult> data = new ArrayList<>();
            boolean finalSuccess = true;
            for (String sql : sqlList) {
                ExecuteSqlResult sqlResult = new ExecuteSqlResult();
                try {
                    catalog.executeSql(tablePath, sql);
                    sqlResult.setSql(sql);
                    sqlResult.setSuccess(true);
                } catch (CatalogException e) {
                    finalSuccess = false;
                    sqlResult.setSql(sql);
                    sqlResult.setMessage(e.getCause().getMessage());
                    sqlResult.setSuccess(false);
                }
                data.add(sqlResult);
            }
            executeResult.setSuccess(finalSuccess);
            executeResult.setResults(data);
        } catch (Exception e) {
            log.error("执行sql出错！", e);
            executeResult.setSuccess(false);
            executeResult.setMessage(e.getMessage());
        }
        return executeResult;
    }
}
